package org.huangrui.spark.java.streaming;

import lombok.val;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

/**
 * @Author hr
 * @Create 2024-10-21 21:19
 */
public class SparkStreaming06_State_Join {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, new Duration(5*1000L));

        val data9999 = jsc.socketTextStream("localhost", 9999);
        val data8888 = jsc.socketTextStream("localhost", 8888);

        JavaDStream<Tuple2<String, Integer>> map9999 = data9999.map(v1 -> new Tuple2<>(v1, 9));
        JavaDStream<Tuple2<String, Integer>> map8888 = data8888.map(v1 -> new Tuple2<>(v1, 8));




        jsc.start();
        jsc.awaitTermination();
    }
}
